package io.dy;

import io.dy.constant.MqConstant;
import io.dy.model.OrderlyModel;
import org.apache.rocketmq.client.consumer.DefaultMQPushConsumer;
import org.apache.rocketmq.client.consumer.listener.ConsumeConcurrentlyContext;
import org.apache.rocketmq.client.consumer.listener.ConsumeConcurrentlyStatus;
import org.apache.rocketmq.client.consumer.listener.MessageListenerConcurrently;
import org.apache.rocketmq.client.exception.MQBrokerException;
import org.apache.rocketmq.client.exception.MQClientException;
import org.apache.rocketmq.client.producer.DefaultMQProducer;
import org.apache.rocketmq.client.producer.MessageQueueSelector;
import org.apache.rocketmq.client.producer.SendCallback;
import org.apache.rocketmq.client.producer.SendResult;
import org.apache.rocketmq.common.message.Message;
import org.apache.rocketmq.common.message.MessageExt;
import org.apache.rocketmq.common.message.MessageQueue;
import org.apache.rocketmq.remoting.exception.RemotingException;
import org.junit.Test;

import java.util.Arrays;
import java.util.Date;
import java.util.List;
import java.util.concurrent.CountDownLatch;

/**
 * @author : dmj
 * @date: 2024/8/23 10:34 上午
 * @description: 原生api
 */
public class RocketMqSimple {

    /**
     * 同步消息
     *
     * @throws Exception
     */
    @Test
    public void producer() throws Exception {

        //1、创建生产者
        DefaultMQProducer producer = new DefaultMQProducer("test-producer-group");

        //2、连接namesrv
        producer.setNamesrvAddr(MqConstant.NAME_SRV_ADDRESS);

        //3、启动
        producer.start();

        //4、发消息
        Message message = new Message("testTopic", "第二条RocketMq消息".getBytes());
        SendResult sendResult = producer.send(message);
        System.out.println(sendResult.getSendStatus());

        //5、关闭
        producer.shutdown();
    }

    /**
     * 异步消息
     *
     * @throws Exception
     */
    @Test
    public void asyncProducer() throws Exception {

        //1、创建生产者
        DefaultMQProducer producer = new DefaultMQProducer("async-producer-group");

        //2、连接namesrv
        producer.setNamesrvAddr(MqConstant.NAME_SRV_ADDRESS);

        //3、启动
        producer.start();

        //4、发消息
        Message message = new Message("asyncTopic", "我是一个异步消息。。。".getBytes());
        producer.send(message, new SendCallback() {

            /**
             * 成功回调
             * @param sendResult
             */
            @Override
            public void onSuccess(SendResult sendResult) {
                System.out.println("发送成功。。。");
            }

            /**
             * 失败回调
             * @param e
             */
            @Override
            public void onException(Throwable e) {
                System.out.println("发送失败:" + e.getMessage());
            }
        });
        System.out.println("主线程执行。。。。");

        new CountDownLatch(1).await();
    }

    /**
     * 单向消息 用于不重要的数据 日志
     */
    @Test
    public void onewayProducer() throws Exception {
        //1、创建生产者
        DefaultMQProducer producer = new DefaultMQProducer("oneway-producer-group");

        //2、连接namesrv
        producer.setNamesrvAddr(MqConstant.NAME_SRV_ADDRESS);

        //3、启动
        producer.start();

        //4、创建消息
        Message message = new Message("onewayTopic", "单向消息".getBytes());
        producer.sendOneway(message);

        System.out.println("成功。。。");
        producer.shutdown();
    }

    /**
     * 延迟消息 比如订单 30分钟后去检查订单状态 若是没有付款 则取消订单释放库存
     */
    @Test
    public void delayedProducer() throws Exception {
        //1、创建生产者
        DefaultMQProducer producer = new DefaultMQProducer("delayed-producer-group");

        //2、连接namesrv
        producer.setNamesrvAddr(MqConstant.NAME_SRV_ADDRESS);

        //3、启动
        producer.start();

        //4、创建消息 并设置延时时间
        Message message = new Message("delayedTopic", "延时消息".getBytes());
        message.setDelayTimeLevel(3);
        producer.send(message);

        System.out.println(new Date());
        producer.shutdown();
    }

    /**
     * 批量消息
     */
    @Test
    public void batchProducer() throws Exception {
        //1、创建生产者
        DefaultMQProducer producer = new DefaultMQProducer("batch-producer-group");

        //2、连接namesrv
        producer.setNamesrvAddr(MqConstant.NAME_SRV_ADDRESS);

        //3、启动
        producer.start();

        //4、创建消息 并设置延时时间
        List<Message> messages = Arrays.asList(
                new Message("batchTopic", "批量消息1".getBytes()),
                new Message("batchTopic", "批量消息2".getBytes()),
                new Message("batchTopic", "批量消息3".getBytes()),
                new Message("batchTopic", "批量消息4".getBytes())
        );
        SendResult send = producer.send(messages);
        System.out.println(send.getSendStatus());
        producer.shutdown();
    }


    public static List<OrderlyModel> msgList = Arrays.asList(
            new OrderlyModel(1, "order01", "下单"),
            new OrderlyModel(1, "order01", "短信"),
            new OrderlyModel(1, "order01", "物流"),

            new OrderlyModel(2, "order02", "下单"),
            new OrderlyModel(2, "order02", "短信"),
            new OrderlyModel(2, "order02", "物流")
    );

    /**
     * 顺序消息 生产者 需要确保消息发到同一个队列中
     */
    @Test
    public void orderlyProducer() throws Exception {
        //1、创建生产者
        DefaultMQProducer producer = new DefaultMQProducer("orderly-producer-group");

        //2、连接namesrv
        producer.setNamesrvAddr(MqConstant.NAME_SRV_ADDRESS);

        //3、启动
        producer.start();

        //4、创建消息并发送
        msgList.forEach(message -> {
            try {
                producer.send(new Message("orderlyTopic", message.toString().getBytes()), new MessageQueueSelector() {

                    /**
                     * @param list list表示topic下的队列数目
                     * @param message 信息
                     * @param obj 此处obj参数 是 send(Message msg, MessageQueueSelector selector, Object arg) 中的arg 回传到方法中
                     * @return 返回具体队列
                     */
                    @Override
                    public MessageQueue select(List<MessageQueue> list, Message message, Object obj) {
                        //obj为message.getOrderNum() 即模型对应的订单号 根据订单号的hash值 对队列大小取模 对应具体的队列序号
                        int hashCode = obj.toString().hashCode() & Integer.MAX_VALUE;
                        int index = hashCode % list.size();
                        System.out.println("hashcode:" + hashCode + "    index:" + index);
                        return list.get(index);
                    }
                }, message.getOrderNum());
            } catch (MQClientException | RemotingException | MQBrokerException | InterruptedException e) {
                e.printStackTrace();
            }
        });

        producer.shutdown();
        System.out.println("发送完成。。。");
    }


//====================================================消费者==================================================================

    /**
     * 简单消费者
     */
    @Test
    public void consumer() throws Exception {

        //1、创建消费者
        DefaultMQPushConsumer consumer = new DefaultMQPushConsumer("test-consumer-group");

        //2、连接namesrv
        consumer.setNamesrvAddr(MqConstant.NAME_SRV_ADDRESS);

        //3、订阅主题    * 表示订阅主题中的所有消息
        consumer.subscribe("testTopic", "*");

        //4、设置监听器
        consumer.registerMessageListener(new MessageListenerConcurrently() {
            /**
             * @param list 消息集合 不过每次都是1条 get(0)
             * @param context 消费上下文
             * @return 返回值 ConsumeConcurrentlyStatus 告诉mq消费成功与否
             */
            @Override
            public ConsumeConcurrentlyStatus consumeMessage(List<MessageExt> list, ConsumeConcurrentlyContext context) {
                System.out.println("我是消费者。。。");
                System.out.println(new String(list.get(0).getBody()));
                System.out.println("消费者上下文:" + context);

                // 返回值 CONSUME_SUCCESS 表示消费成功 消息会从队列中移除
                // RECONSUME_LATER (报错/null) 失败 消息会重新回到队列
                return ConsumeConcurrentlyStatus.CONSUME_SUCCESS;
            }
        });

        //5、启动
        consumer.start();

        //6、阻塞
        new CountDownLatch(1).await();
    }


    /**
     * 延时消费者
     * Fri Aug 23 15:24:15 CST 2024
     * Fri Aug 23 15:24:45 CST 2024
     * <p>
     * Fri Aug 23 15:25:50 CST 2024
     * Fri Aug 23 15:26:00 CST 2024
     */
    @Test
    public void delayedConsumer() throws Exception {

        //1、创建消费者
        DefaultMQPushConsumer consumer = new DefaultMQPushConsumer("delay-consumer-group");

        //2、连接namesrv
        consumer.setNamesrvAddr(MqConstant.NAME_SRV_ADDRESS);

        //3、订阅主题    * 表示订阅主题中的所有消息
        consumer.subscribe("delayedTopic", "*");

        //4、设置监听器
        consumer.registerMessageListener(new MessageListenerConcurrently() {
            /**
             * @param list 消息集合 不过每次都是1条 get(0)
             * @param context 消费上下文
             * @return 返回值 ConsumeConcurrentlyStatus 告诉mq消费成功与否
             */
            @Override
            public ConsumeConcurrentlyStatus consumeMessage(List<MessageExt> list, ConsumeConcurrentlyContext context) {
                System.out.println("接收到消息:" + new Date());
                System.out.println(new String(list.get(0).getBody()));
                // 返回值 CONSUME_SUCCESS 表示消费成功 消息会从队列中移除
                // RECONSUME_LATER (报错/null) 失败 消息会重新回到队列
                return ConsumeConcurrentlyStatus.CONSUME_SUCCESS;
            }
        });

        //5、启动
        consumer.start();

        //6、阻塞
        new CountDownLatch(1).await();
    }

}
